package io.sqrtqiezi.spark.lagou

import org.apache.spark.{SparkConf, SparkContext}

object FindFriends {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    val lines = sc.textFile("lagou-data/fields.dat")
    lines.collect.foreach(println)

    val friendRDD = lines.map { line =>
      val fields = line.split(",")
      val userId = fields(0).trim
      val friends = fields(1).trim.split("\\s+")
      (userId, friends)
    }
    friendRDD.collect.foreach(println)

    println("--------------------------------------------------")
    friendRDD.flatMapValues(friends => friends.combinations(2))
      .map{ case (k, v) => (v.mkString(" & "), Set(k))}
      .reduceByKey(_ | _)
      .sortByKey()
      .collect.foreach(println)
  }
}
